package com.bst.datax.executor.util;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHeaders;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

public class StreamLoadTest {
    public static void main(String[] args) throws IOException {
        List<Map<String,Object>> datas = loadData();
        String label = "label_user_java";
        String username = "root";
        String password = "";
        String db = "demo";
        String table ="user";
        String loadUrl = String.format("http://192.168.179.131:8030/api/%s/%s/_stream_load",db,table);
        List<String> columns = Arrays.asList("id,username,age,sex,phone,register_time".split(","));
        String columnSeparator = ",";
        String format = "csv";
        String  loadData = datas.stream().map(data -> columns.stream().map(column -> data.get(column).toString()).collect(Collectors.joining(columnSeparator))).collect(Collectors.joining("\n"));
        sendData(label,username,password,loadUrl,columns,loadData,columnSeparator,null,format,null);
    }

    /**
     * 加载数据
     * @return
     */
    private static List<Map<String, Object>> loadData() {
        List<Map<String,Object>> datas = new ArrayList<>();
        Map<String,Object> map = new HashMap<>();
        map.put("id",6);
        map.put("username","user_6");
        map.put("age",52);
        map.put("sex",1);
        map.put("phone","13612345678");
        map.put("register_time","2023-11-02-12:34:36");
        datas.add(map);
        return  datas;
    }

    /**
     * Basic access authentication 签名
     * @param username doris用户名
     * @param password doris用户密码
     * @return
     */
    public static String basicAuthHeader(String username, String password) {
        final String tobeEncode = username + ":" + password;
        byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8));
        return "Basic " + new String(encoded);
    }
    /**
     * Stream load 导入数据
     * @param label 导入任务的标识。每个导入任务，都有一个在单 database 内部唯一的 label。label 是用户在导入命令中自定义的名称。通过这个 label，用户可以查看对应导入任务的执行情况。
     * @param username
     * @param password
     * @param loadUrl
     * @param columns 待导入数据的函数变换配置，目前 Stream load 支持的函数变换方法包含列的顺序变化以及表达式变换，其中表达式变换的方法与查询语句的一致。
     * @param loadData
     * @param columnSeparator 用于指定导入文件中的列分隔符，默认为\t。如果是不可见字符，则需要加\x作为前缀，使用十六进制来表示分隔符。
     * @param lineDelimiter 用于指定导入文件中的换行符，默认为\n。
     * @param format 指定导入数据格式，支持csv、json，默认是csv
     * @param mergeType 数据的合并类型:一共支持三种类型APPEND、DELETE、MERGE 其中，APPEND是默认值
     * @throws IOException
     */
    public static void sendData(String label, String username, String password, String loadUrl
            , List<String> columns, String loadData, String columnSeparator, String lineDelimiter
            , String format, String mergeType) throws IOException {
        HttpClientBuilder
                httpClientBuilder = HttpClients
                .custom()
                .setRedirectStrategy(new DefaultRedirectStrategy() {
                    @Override
                    protected boolean isRedirectable(String method) {
                        return true;
                    }
                });
        try (CloseableHttpClient client = httpClientBuilder.build()) {
            HttpPut put = new HttpPut(loadUrl);
            StringEntity entity = new StringEntity(loadData, "UTF-8");
            put.setHeader(HttpHeaders.EXPECT, "100-continue");
            put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(username, password));
            // the label header is optional, not necessary
            // use label header can ensure at most once semantics
            put.setHeader("label", label);
            if(StringUtils.isNotBlank(columnSeparator)){
                put.setHeader("column_separator", columnSeparator);
            }

            if(StringUtils.isNotBlank(lineDelimiter)){
                put.setHeader("line_delimiter", lineDelimiter);
            }


            put.setHeader("format", format);

            put.setHeader("merge_type", mergeType);
            //字段
            if (null != columns && !columns.isEmpty()) {
                put.setHeader("columns", String.join(",",
                        columns.stream().map(f -> String.format("`%s`", f)).
                                collect(Collectors.toList())));
            }
            //数据
            put.setEntity(entity);

            try (CloseableHttpResponse response = client.execute(put)) {
                String loadResultStr  =  null;
                if (response.getEntity() != null) {
                    loadResultStr  =  EntityUtils.toString(response.getEntity());
                }
                final int statusCode = response.getStatusLine().getStatusCode();
            }
        }
    }
}
